package com.djl.Zookeeper.lock;

import com.djl.Zookeeper.util.ZkUtil;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 锁管理
 */
public class ZkLock implements Watcher, AsyncCallback.StringCallback, AsyncCallback.ChildrenCallback {

    private static int i = 0;

    private static ZooKeeper zk;

    private String threadName;

    private String lockPath;

    private CountDownLatch downLatch = new CountDownLatch(1);


    /**
     * 获取锁
     */
    public void tryLock() {
        zk = ZkUtil.getAloneZookeeper();
        try {
            //方便观察线程创建节点乱序的现象
            String threadName = Thread.currentThread().getName();
            this.threadName = threadName;
            //创建前缀为lock的临时序列节点
            zk.create("/lock", threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,this,"lock");
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 释放锁
     */
    public void unLock() throws KeeperException, InterruptedException {
        System.out.println("当前释放锁的路径："+lockPath+" .当前线程为："+ threadName);

        Stat stat = new Stat();
        zk.getData(lockPath,false,stat);
        // version还可以填-1,表示忽略版本
        zk.delete(lockPath,stat.getVersion());
        downLatch = new CountDownLatch(1);
    }



    /**
     *  创建完成的回调 zk.create
     *  判断是否是最小的哪一个序列节点,如果是则获取锁
     */
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name!=null){
            //System.out.println(threadName+" create Node : "+name);
            //方便后边判断自己是否是第一个
            lockPath = name;
            // 此时不需要监控父节点 "/lockNode" , 只需要监控子节点，只是通过父节点获取子节点
            zk.getChildren("/",false,this,"lock");
        }
    }

    /**
     * 获取所有子节点的回调 zk.getChildren
     */
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children) {

        //先排序
        Collections.sort(children);

        int index = children.indexOf(lockPath.substring(1));
        //判断自己是否是第一个
        if(index==0){
            i++;
            System.out.println(lockPath+"是第"+i+"个");
            downLatch.countDown();
        }else{
            //第一种：如果不是第一个就 getChildren 但是这样的效率极其低下
            //zk.getChildren("/",false,this,"lock");

            //第二种：剩下每个线程监控自己的上个节点
            try {
                zk.exists("/"+children.get(index-1),this);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 事件监听回调
     * zk 连接 事件回调监听
     * @param event
     */
    @Override
    public void process(WatchedEvent event) {
        Event.KeeperState state = event.getState();
        Event.EventType type = event.getType();
        switch (type) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                //当上个节点删除了
                zk.getChildren("/",false,this,"lock");
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            case DataWatchRemoved:
                break;
            case ChildWatchRemoved:
                break;
            case PersistentWatchRemoved:
                break;
        }
        switch (state) {
            case Unknown:
                break;
            case Disconnected:
                break;
            case NoSyncConnected:
                break;
            case SyncConnected:
                break;
            case AuthFailed:
                break;
            case ConnectedReadOnly:
                break;
            case SaslAuthenticated:
                break;
            case Expired:
                break;
            case Closed:
                break;
        }
    }


    public String getThreadName() {
        return threadName;
    }

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public String getLockPath() {
        return lockPath;
    }

    public void setLockPath(String lockPath) {
        this.lockPath = lockPath;
    }

}
